package com.dongdongshop.consumer;

import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.dongdongshop.pojo.TbSeckillGoods;
import com.dongdongshop.pojo.TbSeckillOrder;
import com.dongdongshop.service.SeckillOrderService;

import com.dongdongshop.util.IdWorker;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.Date;
import java.util.List;

/**
 * @Description
 * @ClassName ScoreMQConsumer
 * @Author RC
 * @date 2020.12.27 17:58
 */
@Component
@RocketMQMessageListener(consumerGroup = "seckill_consumer",topic = "seckill")
public class SeckillMQConsumer implements RocketMQListener<String>,
        RocketMQPushConsumerLifecycleListener {

    @Autowired
    private RedisTemplate redisTemplate;
    @Reference(timeout = 100000,check = false)
    private SeckillOrderService seckillOrderService;
    private Logger logger = LoggerFactory.getLogger(SeckillMQConsumer.class);

    @Override
    public void onMessage(String message) {

    }

    /**
     * 自定义消息消费重试策略（手动接管ACK确认机制）
     * CONSUME_SUCCESS、RECONSUME_LATER
     * 注意：保留原有消费者监听RocketMQListener
     */
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                MessageExt msg = list.get(0);
                String msgbody = null;
                try {
                    msgbody = new String(msg.getBody(), "utf-8");
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }

                JSONObject jsonObject = JSON.parseObject(msgbody);
                Long seckillId = jsonObject.getObject("seckillId", Long.class);
                String username = jsonObject.getObject("username", String.class);
                try {
                    //解决幂等性问题 消费之前我们要判断这条消息有没有被消费成功
                    //如果成功消费后 将消息id最为key保存到redis中

                    //String msgbody = new String(msg.getBody(), "utf-8");



                    TbSeckillGoods seckillGoods =(TbSeckillGoods) redisTemplate.boundHashOps("seckillGoods").get(seckillId);

                    // 不是最后一件商品  就下单
                    TbSeckillOrder seckillOrder = new TbSeckillOrder();
                    IdWorker idWorker = new IdWorker();
                    seckillOrder.setId(idWorker.nextId());
                    seckillOrder.setCreateTime(new Date());
                    seckillOrder.setMoney(seckillGoods.getCostPrice());
                    seckillOrder.setSeckillId(seckillId);
                    seckillOrder.setSellerId(seckillGoods.getSellerId());
                    seckillOrder.setStatus("0");
                    seckillOrder.setUserId(username);
                    //保存订单
                    seckillOrderService.createOrder(seckillOrder);
                    //订单保存成功记录缓存
                    redisTemplate.boundHashOps("seckillOrder").put(username,seckillOrder);

                }catch (Exception e) {
                    //e.printStackTrace();
                    logger.error("消息失败");
                    // 该条消息可以存储到DB或者LOG日志中，或其他处理方式，然后重试
                    //先获取消息的iD 消息的重试次数
                    int reconsumeTimes = msg.getReconsumeTimes();
                    logger.info("消息重试次数 为:" + reconsumeTimes);
                    if(reconsumeTimes >= 2 ){
                        //整合redis 将这条消息保存到死信队列中,(用redis充当死信队列) 当然也可以保存到mysql
                      logger.info("已经消费三次了 不再重试 ,将数据保存到私信队列中");

                        TbSeckillOrder seckillOrder = new TbSeckillOrder();
                        seckillOrder.setId(-999L);
                        redisTemplate.boundHashOps("seckillOrder").put(username,seckillOrder);
                        redisTemplate.boundValueOps("stockCount_" + seckillId).increment();
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                   logger.info("消息的id 为 :" + msg.getMsgId());

                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
                }
                //返回之前讲消息id保存到redis中
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
